package com.example.KafkaDdemo.config;

import com.example.KafkaDdemo.filter.ConsumerinterceptorTTL;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private Boolean autoCommit;

    @Value("${spring.kafka.consumer.properties.group.id}")
    private String groupId;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeserializer;
    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeserializer;
    @Value("${spring.kafka.bootstrap-servers}")
    private String borker;

    /**
     * 消费者配置信息
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<String, Object>();
//        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 手动提交还是自动提交
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, borker);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);// 手动提交 配置 false
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        // max.poll.records:表示批量拉取的情况下，一次最多拉取多少个数据
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
        // max.poll.interval.ms:表示消费者活跃检测机制，如果调用poll()的频率大于这个值，这个消费者提出消费者组
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 6000);
        // 指定消费者拦截器
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ConsumerinterceptorTTL.class.getName());
        return props;
    }

    /**
     * 消费者批量工程
     */
    @Bean(name = "simpleFactory")
    public KafkaListenerContainerFactory<?> batchFactory() {

        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        // 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        /**
         * setConcurrency详解
         * 如果没有设置
         *  KafkaMessageListenerContainer由spring提供用于监听以及拉取消息，并将这些消息按指定格式转换后交给由@KafkaListener注解的方法处理，
         *  相当于一个消费者，其入口为doStart方法，内部会创建一个ListenerConsumer，并交给线程池处理，ListenerConsumer放到线程池之后，会调用
         *  自身的run方法，run方法内部逻辑是while循环不断拉取消息，并丢给@KafkaListener注解对应的方法。
         *
         * 如果设置了
         *  就会使用ConcurrentMessageListenerContainer，其底层逻辑仍然是通过KafkaMessageListenerContainer实现处理；
         *  从实现上看就是在KafkaMessageListenerContainer上做了层包装，
         *  有多少的concurrency就创建多个KafkaMessageListenerContainer，也就是concurrency个消费者
         *
         *  假如我们topic有三个分区，但是我们服务器只有一台，那么我们为了提高消费效率，想有三个消费者，就可以把这个参数设置为3
         *  如果我们有三台服务器，就设置为1.
         */
        factory.setConcurrency(3);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        factory.getContainerProperties().setPollTimeout(30000);
        return factory;
    }


}
